package com.guass.nettyprotostuff.service.impl;

import com.guass.nettyprotostuff.rpc.common.Session;
import com.guass.nettyprotostuff.rpc.common.ModelMsg;
import com.guass.nettyprotostuff.rpc.common.SessionManager;
import com.guass.nettyprotostuff.rpc.common.TypeData;
import com.guass.nettyprotostuff.service.MqService;
import com.guass.nettyprotostuff.service.RpcService;
import com.guass.nettyprotostuff.zookeeper.ZookeeperService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.stream.Collectors;


/**
 * @author guass
 */
@Service
@Slf4j
public class RpcServiceImpl implements RpcService {

    private final SessionManager sessionManager = SessionManager.getInstance();

    @Resource
    private ZookeeperService zookeeperService;

    @Resource
    private MqService mqService;

    @Override
    public void onDataReceive(ChannelHandlerContext ctx, ModelMsg msg) {

        log.info("guass 收到客户端发送数据 onDataReceive {},{}", msg.getBody(), ctx.channel().id().asShortText());
        mqService.sendAsyncMessage("收到客户端发送数据 ==> " + msg.getBody() + " client id ==> " + ctx.channel().id().asShortText());
    }

    @Override
    public void onTerminalActive(ChannelHandlerContext ctx) {

        Session session = Session.buildSession(ctx.channel());
        sessionManager.put(session.getId(), session);
        log.info("guass 终端连接 {}, {}",session);

    }

    @Override
    public void onTerminalInActive(ChannelHandlerContext ctx) {


        final String sessionId = ctx.channel().id().asLongText();
        Session session = sessionManager.findBySessionId(sessionId);
        log.info("终端断开连接:{}", session);
        this.sessionManager.removeBySessionId(sessionId);
        ctx.channel().close();
    }

    @Override
    public void sendData(Channel channel, ModelMsg modelMsg) {
        channel.writeAndFlush(modelMsg);
    }

    @Override
    public void sendMessageToAll(String msg) {
        ModelMsg modelMsg = new ModelMsg();

        modelMsg.setMessageType(TypeData.CUSTOMER);
        modelMsg.setMsgId("guass server");
        modelMsg.setBody(msg);


        List<Session> sessions = sessionManager.getSessions();

        log.info("guass sessionMap {}",sessions.size());
        sessions.forEach(item ->{
            item.getChannel().writeAndFlush(modelMsg);
        });
    }

    @Override
    public int getChannelCount() {
        return sessionManager.getChannelCount();
    }

    private void sendMessageToAll(String msg,Channel exChannel){
        ModelMsg modelMsg = new ModelMsg();

        modelMsg.setMessageType(TypeData.CUSTOMER);
        modelMsg.setMsgId("guass server");
        modelMsg.setBody(msg);


        List<Session> sessions = sessionManager.getSessions();

        sessions = sessions.stream().filter( item ->
                !item.getChannel().id().asLongText().equals(exChannel.id().asLongText())
        ).collect(Collectors.toList());

        log.info("guass {}",sessions);
        sessions.forEach(item ->{
            Channel channel = item.getChannel();

            if (channel.isActive() && channel.isWritable()) {
                item.getChannel().writeAndFlush(modelMsg);
            }

        });
    }
}
